
#include "rtmp_client.h"
#include "rtmp_connection.h"
#include "rtmp_message.h"
#include <cstddef>
#include <cstdint>
#include <cstdlib>
#include <functional>

#include <memory>
#include <string>
#include <vector>


RtmpClient::RtmpClient(Loop * loop)
{
    m_loop = loop;
}

RtmpClient::~RtmpClient()
{
    m_client = nullptr;
}

bool RtmpClient::ConnectionServer(const std::string & url)
{
    bool ret = ParseRtmpUrl(url);
    if (!ret)
    {
        return false;
    }
    m_client = std::make_shared<Client>(m_loop);
    m_client->MessageCallbackFunc(std::bind(&RtmpClient::OnMessage, this, std::placeholders::_1));
    m_client->ErrorCallbackFunc(std::bind(&RtmpClient::OnError, this, std::placeholders::_1));
    ClientConfig config;
    config.socket_block = false;
    /* bool result = m_client->Connect(url, config); */
    ret = m_client->Connect(m_url_info.m_domain, m_url_info.m_port, ClientConfig{});
    if (ret)
    {
        m_conn = std::make_shared<RtmpConnection>(m_client, true);
        if (m_loop)
        {
            m_loop->AddPendingTask([this](){
                    
                INFO("execute pending task and begin start pull stream");
                StartPullStream();
            });
        }
    }
    return ret;
}

void RtmpClient::StartPullStream()
{
    m_status_func[EmRtmpConnectInitFinish].push_back(std::bind(&RtmpClient::SendConnect, this));
    m_status_func[EmRtmpConnectConnectCmd].push_back(std::bind(&RtmpClient::SendCreateStream, this));
    m_status_func[EmRtmpConnectCreateStreamCmd].push_back(std::bind(&RtmpClient::SendGetStreamLength, this));
    m_status_func[EmRtmpConnectCreateStreamCmd].push_back(std::bind(&RtmpClient::SendPlay, this));
    SendC0C1();
}

bool RtmpClient::ParseRtmpUrl(const std::string & url)
{
    auto index = url.find("://");
    if (index == std::string::npos)
    {
        return false;
    }
    std::string protocol = url.substr(0, index);
    std::string domain;
    std::string apps;
    std::string streams;
    uint16_t port = 1935;
    auto port_index = url.find(":", index + 3);
    if (port_index == std::string::npos)
    {
        if (protocol == "rtmp")
        {
            // default port is 1935
        }
        else if (protocol == "rtmpt")
        {
            // default port is 80
        }
        else if (protocol == "rtmpe")
        {
            // default port is 1935
        }
        else if (protocol == "rtmps")
        {
        
            // pprotocol port is 1935
        }
        else if (protocol == "rtmpte")
        {
            // protocol port is 80 
        }
        else if (protocol == "rtmpts")
        {
            // protocol port is 443 
        }
        port = 1935;
        std::size_t app_index = url.find("/", index + 3);
        if (app_index == std::string::npos)
        {
            return false;
        }
        std::size_t stream_index = url.find("/", app_index + 1);
        if (stream_index == std::string::npos)
        {
            return false;
        }
        domain = url.substr(index + 3, app_index - index - 3);
        apps = url.substr(app_index + 1, stream_index - app_index - 1);
        streams = url.substr(stream_index + 1);
    }
    else
    {
        std::size_t app_index = url.find("/", port_index + 1);
        if (app_index == std::string::npos)
        {
            return false;
        }
        std::size_t stream_index = url.find("/", app_index + 1);
        if (stream_index == std::string::npos)
        {
            return false;
        }
        domain = url.substr(index + 3, port_index - index - 3);
        port = atoi(url.substr(port_index + 1, app_index - port_index).c_str());
        apps = url.substr(app_index + 1, stream_index - app_index - 1);
        streams = url.substr(stream_index + 1);
    }
    INFO("protocol : {} domain : {} port : {} app : {} stream : {}", protocol, domain, port, apps, streams);
    m_url_info.m_protocol = protocol;
    m_url_info.m_domain   = domain;
    m_url_info.m_port     = port;
    m_url_info.m_app      = apps;
    m_url_info.m_stream   = streams;
    return true;
}

void RtmpClient::OnMessage(std::shared_ptr<SocketBuffer> socket)
{
    const uint8_t * data = socket->GetBufferData();
    const int length     = socket->GetBufferLength();
    int use_length = 0;
    if (m_conn)
    {
        use_length = m_conn->PushRtmpData(data, length);
        socket->FreeBufferData(use_length);
    }
    ChangeTask();
}

void RtmpClient::UpdateStatus(int status)
{
    if (m_conn)
    {
        m_conn->UpdateState((RtmpConnectState)status);
    }
}


void RtmpClient::OnError(std::string msg)
{

}

void RtmpClient::ChangeTask()
{
    if (m_conn)
    {
        auto state = m_conn->State();
        auto task_iter = m_status_func.find(state);
        if (task_iter != m_status_func.end())
        {
            for (auto & func : task_iter->second)
            {
                func();
            }
        }
    }
}

void RtmpClient::SendC0C1()
{
    std::vector<uint8_t> data(1537);
    data[0] = 0x03;
    for (int index = 1; index < 1536; index++)
    {
        data[index] = rand() % 10;
    }
    if (m_conn)
    {
        m_conn->SendData(&*data.begin(), data.size());
        m_conn->UpdateState(EmRtmpConnectSendC0C1);
    }
}

void RtmpClient::SendConnect()
{
    std::shared_ptr<RtmpMessage> connect_msg = std::make_shared<RtmpMessage>();
    connect_msg->m_header.m_format  = 0x00;
    connect_msg->m_header.m_chunk_stream = 0x03;
    connect_msg->m_header.m_timestamp = 0;
    connect_msg->m_header.m_calc_timestamp = 0;
    connect_msg->m_header.m_type_id = 0x14;
    connect_msg->m_header.m_stream_id = 0;
	
    auto cmd = new AmfString();
    cmd->SetString("connect");
    auto num = new AmfNumber();
    num->SetValue(1);
    auto content = new AmfObject();
    content->SetString("app", m_url_info.m_app);
    content->SetString("flashVer", "LNX 9,0,124,2");
    std::string tc_url = m_url_info.m_protocol + "://" + m_url_info.m_domain + ":" + std::to_string(m_url_info.m_port) + "/" + m_url_info.m_app + "/" + m_url_info.m_stream;
    content->SetString("tcUrl", tc_url);
    content->SetBool("fpad", false);
    content->SetNumber("capabilities", 15);
    content->SetNumber("audioCodecs",  4071);
    content->SetNumber("videoCodecs",  252);
    content->SetNumber("videoFunction",1);

    connect_msg->PushAmfValue(cmd);
    connect_msg->PushAmfValue(num);
    connect_msg->PushAmfValue(content);

    SendMsg(connect_msg);
    UpdateStatus(EmRtmpConnectConnectCmd);
}

void RtmpClient::SendCreateStream()
{
    auto create_msg = std::make_shared<RtmpMessage>();
    create_msg->m_header.m_format = 0x01;
    create_msg->m_header.m_chunk_stream   = 0x03;
    create_msg->m_header.m_timestamp      = 0;
    create_msg->m_header.m_calc_timestamp = 0;
	create_msg->m_header.m_type_id        = 0x14;
	
    auto cmd = new AmfString();
    cmd->SetString("createStream");

    auto num = new AmfNumber();
    num->SetValue(2);

    auto content = new AmfNull();

    create_msg->PushAmfValue(cmd);
    create_msg->PushAmfValue(num);
    create_msg->PushAmfValue(content);

    SendMsg(create_msg);
    UpdateStatus(EmRtmpConnectCreateStreamCmd);
}

void RtmpClient::SendGetStreamLength()
{
    auto msg = std::make_shared<RtmpMessage>();

    msg->m_header.m_format = 0x01;
    msg->m_header.m_chunk_stream = 0x03;
    msg->m_header.m_timestamp    = 0;
    msg->m_header.m_calc_timestamp= 0;
    msg->m_header.m_type_id      = 0x14;

    auto cmd = new AmfString();
    cmd->SetString("getStreamLength");

    auto number = new AmfNumber();
    number->SetValue(3);

    auto null = new AmfNull();

    auto stream = new AmfString();
    stream->SetString(m_url_info.m_stream);

    msg->PushAmfValue(cmd);
    msg->PushAmfValue(number);
    msg->PushAmfValue(null);
    msg->PushAmfValue(stream);

    SendMsg(msg);
}

void RtmpClient::SendPlay()
{
    auto play = std::make_shared<RtmpMessage>();

    play->m_header.m_format = 0x01;
    play->m_header.m_chunk_stream = 0x03;
    play->m_header.m_timestamp    = 0;
    play->m_header.m_calc_timestamp= 0;
    play->m_header.m_type_id      = 0x14;
    auto cmd = new AmfString();
    cmd->SetString("play");

    auto number = new AmfNumber();
    number->SetValue(4);

    auto null = new AmfNull();

    auto stream = new AmfString();
    stream->SetString("stream");

    auto number_time = new AmfNumber();
    number_time->SetValue(-2000);

    play->PushAmfValue(cmd);
    play->PushAmfValue(number);
    play->PushAmfValue(null);
    play->PushAmfValue(stream);
    play->PushAmfValue(number_time);

    SendMsg(play);
    UpdateStatus(EmRtmpConnectPlayCmd);
}

void RtmpClient::SendMsg(std::shared_ptr<RtmpMessage> & msg)
{
    if (m_conn && msg)
    {
        m_conn->SendRespMsg(msg);
    }
}
